package aa;

import java.nio.channels.SocketChannel;
import java.util.*;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.lang.Math;
import java.lang.Thread;
import java.lang.reflect.Array;

class myThr extends Thread {

	streamc st;

	public void run() {
		try {

			st.doWork();
		} catch (Exception e) {

			helpFunc.print(e.toString());
		}
	}
}

public class streamc extends streambase {
	Map<DatagramChannel, sockMapData> sockMap = new HashMap<DatagramChannel, sockMapData>();
	Map<Integer, Integer> availPort = new HashMap<Integer, Integer>();
	Map<Integer, Integer> availPort2 = new HashMap<Integer, Integer>();

	LinkedList<Integer> cachePort = new LinkedList<Integer>();
	byte[] salt;
	String ip;
	int statisOut, statisGot;
	Map<Integer, DatagramChannel> reusedPort = new HashMap<Integer, DatagramChannel>();
	double newPortLimit;
	int newPortThisPeriod = 0, newPortThisSecond = 0;
	double maxRecTime;
	LinkedList<Double> maxRecTimeQ = new LinkedList<Double>();
	double minRecTime = Double.MAX_VALUE;
	Map<Double, Integer> newPortMap = new HashMap<Double, Integer>();
	int decreaseDose = 0;
	double startTime;

	double timeoutTime = helpFunc.timeoutTime;
	boolean tooMuchPorts1 = false, tooMuchPorts2 = false;
	logClass myLog;
	int con_portIncreaseDose,speedLimit,perSecondPortsLimit,con_portDecreaseDose,maxPortNum;
	
	
	public streamc(String serverip, int increaseDose,int deDose,int pushA,
			 int sizeLi,int limit,int speed,ArrayList<Integer> lp,byte[] salt_a,logClass lo)
	
			throws Exception {
		
		super( pushA, sizeLi, false);
		maxPortNum = lp.size();
		myLog = lo;
		salt = salt_a;
		for (int i : lp) {
			if (availPort.size() >= helpFunc.minPortNum) {
				cachePort.add(i);
				continue;
			}
			availPort.put(i, 0);
		}
		ip = serverip;
		statisGot = statisOut = 0;
		newPortLimit = helpFunc.regulateTimeSpan * limit;
		for (int i = 0; i < 20; i++)
			maxRecTimeQ.add((double) 0);
		startTime = helpFunc.getTime();
		 speedLimit = speed;
		 con_portIncreaseDose = increaseDose;
		perSecondPortsLimit = limit;
		con_portDecreaseDose = deDose;
	}

	int calPara() {
		if (statisGot != 0)
			waitingTime = maxRecTime - minRecTime;
		int dose = 0;
		if (rRaw > speedLimit * 1024) {
			if (statisOut > helpFunc.speedLimit_portNum)
				dose = -1;
		} else
			dose = con_portIncreaseDose;
		newPortLimit = helpFunc.regulateTimeSpan * perSecondPortsLimit;
		if (statisGot < helpFunc.minPackGot)
			sendStatusRate = helpFunc.con_maxRate
					- (helpFunc.con_maxRate - helpFunc.con_minRate) * ((double) statisGot / helpFunc.minPackGot);
		else
			sendStatusRate = helpFunc.con_minRate;
		if (statisGot == 0)
			timeoutTime = helpFunc.timeoutTime;
		else
			timeoutTime = Collections.max(maxRecTimeQ) + 0.1;
		double lossRate = 1;
		if (statisGot + statisOut != 0)
			lossRate = (double) statisOut / (statisGot + statisOut);

		if (lossRate >= 0 && lossRate <= 0.05) {
			slope = 1;
			maxSend = 3;
		} else if (lossRate >= 0.05 && lossRate <= 0.1) {
			slope = 1.5;
			maxSend = 3;
		} else if (lossRate >= 0.1 && lossRate <= 0.4) {
			slope = 2;
			maxSend = 4;
		} else if (lossRate >= 0.4 && lossRate <= 0.6) {
			slope = 3;
			maxSend = 5;
		} else if (lossRate >= 0.6 && lossRate <= 0.8) {
			slope = 4;
			maxSend = 8;
		} else if (lossRate >= 0.8 && lossRate <= 1) {
			slope = 5;
			maxSend = 10;
		}
		return dose;
	}

	void calNewPortThisPeriod() {
		double t = helpFunc.getTime();
		ArrayList<Double> temp = new ArrayList<Double>(newPortMap.keySet());
		for (double k : temp) {
			if (t - k > helpFunc.regulateTimeSpan)
				newPortMap.remove(k);
		}
		if (newPortMap.size() == 0)
			newPortThisPeriod = 0;
		else {
			newPortThisPeriod = newPortMap.values().stream().mapToInt(Integer::intValue).sum();
		}
	}

	double refreshNewPortTime() {
		double t = helpFunc.getTime();
		if (newPortMap.isEmpty() || newPortMap.values().stream().mapToInt(Integer::intValue).sum() < newPortLimit)
			return t;
		int su = newPortMap.values().stream().mapToInt(Integer::intValue).sum();
		ArrayList<Double> temp = new ArrayList<Double>(newPortMap.keySet());
		Collections.sort(temp);
		for (double i : temp) {
			su -= newPortMap.get(i);
			if (su < newPortLimit)
				return i + helpFunc.regulateTimeSpan;
		}
		return t;
	}

	double refreshNextTimeout() {
		double t = helpFunc.getTime();
		if (sockMap.isEmpty())
			return Double.MAX_VALUE;
		double minT = t;
		for (DatagramChannel k : sockMap.keySet()) {
			sockMapData v = sockMap.get(k);
			if (v.createTime < minT)
				minT = v.createTime;
		}
		return minT + timeoutTime;

	}

	void adjustPortNum(int dose) {
		if (tooMuchPorts1 && !tooMuchPorts2) {
			decreaseDose = con_portDecreaseDose;
			return;
		}
		if (blankRec != 0 && blankSend != 0 && statisGot > helpFunc.minPackGot) {
			decreaseDose = con_portDecreaseDose;
			return;
		}
		if (dose < 0) {
			decreaseDose = -1 * dose;
			return;
		}
		for (int i = 0; i < dose; i++) {
			if (cachePort.size() > 0) {
				int n = cachePort.getFirst();
				cachePort.removeFirst();
				availPort.put(n, 0);
			}
		}
	}

	public void doWork() throws Exception {
		makeOutSo();
		while (true) {
			doWork2();
		}
	}

	public void doWork2() throws Exception {
		if (helpFunc.getTime() - updatedTime > helpFunc.con_close)
			System.exit(0);
		double t = helpFunc.getTime();
		double t1;

		if (availPort.size() != 0) {
			t1 = refreshNewPortTime();
			tooMuchPorts1 = true;
		} else {
			t1 = Double.MAX_VALUE;
			tooMuchPorts2 = true;
		}

		double t2 = refreshNextTimeout();
		double mTime = Math.min(t1, t2);
		double wt;
		if (mTime == Double.MAX_VALUE)
			wt = 1;
		else if (t > mTime) {
			wt = 0;
		} else {
			wt = mTime - t;
		}
		if (sockMap.isEmpty())
			Thread.sleep((long) (wt * 1000));

		else {
			int outSoSign = 0;
			boolean doR = false;
			boolean doW = false;
			if (writeBuffer.length < helpFunc.con_streamBufferSize) {
				outSoSign |= SelectionKey.OP_READ;
				doR = true;
			}
			if (readBuffer.length > 0) {
				outSoSign |= SelectionKey.OP_WRITE;
				doW = true;
			}
			Selector selector = Selector.open();
			if (doR || doW)
				outSo.register(selector, outSoSign);
			for (DatagramChannel channel : sockMap.keySet())
				channel.register(selector, SelectionKey.OP_READ);

			ArrayList<SelectableChannel> rlR = new ArrayList<SelectableChannel>();
			ArrayList<SelectableChannel> wlR = new ArrayList<SelectableChannel>();

			int ret = selector.select(1000);
			Iterator selectedKeys = selector.selectedKeys().iterator();
			while (selectedKeys.hasNext()) {
				SelectionKey key = (SelectionKey) selectedKeys.next();
				selectedKeys.remove();
				if (!key.isValid()) {
					continue;
				}
				if (key.channel() == outSo && key.isReadable() && doR)
					rlR.add(key.channel());
				if (key.channel() == outSo && key.isWritable() && doW)
					wlR.add(key.channel());
				if (key.channel() != outSo && key.isReadable())
					rlR.add(key.channel());
			}
			if (!wlR.isEmpty()) {
				ByteBuffer temp = ByteBuffer.wrap(readBuffer);
				outSo.write(temp);
				readBuffer = new byte[temp.remaining()];
				temp.get(readBuffer, 0, readBuffer.length);

			}
			if (rlR.contains(outSo)) {
				ByteBuffer buf = ByteBuffer.allocate(helpFunc.con_streamBufferSize - writeBuffer.length);

				int bytesRead = outSo.read(buf);
				byte[] temp = new byte[writeBuffer.length + bytesRead];
				System.arraycopy(writeBuffer, 0, temp, 0, writeBuffer.length);						
				System.arraycopy(buf.array(), 0, temp, writeBuffer.length, bytesRead);
				writeBuffer = temp;
			}
			ArrayList<byte[]> re = deal_rec(rlR);
			deal_data_back(re);
			selector.close();
		}
		read();
		write();
		deal_timeout();
		calNewPortThisPeriod();
		int l = availPort.size();

		if (l + newPortThisPeriod > newPortLimit) {
			l = (int)( newPortLimit - newPortThisPeriod);
			if (l < 0)
				l = 0;
		}
		int sendNum = l + availPort2.size();

		ArrayList<byte[]> re = get_data_to_send(sendNum);
		sendData(re);
		if (helpFunc.getTime() - staTime > 1) {
			staTime = helpFunc.getTime();

			String log = String.format("%3d %4d %4d|%4d %4d <s,r>%5d %5d %5d %5d",
					sockMap.size(),	statisGot,statisOut,(int)(rRaw/1024),(int)(wRaw/1024),
					helpFunc.getPackStaBigV(maxSendL),helpFunc.getPackStaBigV(peerMaxRec),
					helpFunc.getPackStaBigV(peerMaxSend),helpFunc.getPackStaBigV(maxRec));
					

			
			myLog.setLog(log);

			newPortThisSecond = 0;
			maxRecTimeQ.add(maxRecTime);
			maxRecTimeQ.removeFirst();

			int dose = calPara();
			adjustPortNum(dose);
			
			tooMuchPorts1 = tooMuchPorts2 = false;
			statisGot = statisOut = 0;
			maxRecTime = 0;
			minRecTime = Double.MAX_VALUE;
			rRaw = wRaw = rNet = wNet = 0;
			totalRec = blankRec = totalSend = blankSend = statusRev = statusSend = 0;
			helpFunc.clearPackSta(maxSendL);
			helpFunc.clearPackSta(maxRec);
			helpFunc.clearPackSta(peerMaxRec);
			helpFunc.clearPackSta(peerMaxSend);
		}
	}
	boolean retireSock(int n) {
		if (cachePort.size() >= maxPortNum - helpFunc.minPortNum || decreaseDose == 0)
			return false;
		decreaseDose -= 1;
		if (availPort.size() > 0) {
			ArrayList<Integer> n2 = new ArrayList<Integer>(availPort.keySet());
			availPort.remove(n2.get(0));
			cachePort.add(n2.get(0));
			return false;
		}
		cachePort.add(n);
		return true;
	}
	void sendData(ArrayList<byte[]> re) throws Exception {
		int co = -1;
		int l = re.size();
		double ft = helpFunc.getTime();
		ArrayList<Integer> tt = new ArrayList<Integer>(availPort2.keySet());
		tt.addAll(availPort.keySet());
		for (int k : tt) {
			co += 1;
			if (co == l)
				return;
			byte[] data = re.get(co);
			DatagramChannel sock;
			if (reusedPort.containsKey(k)) {
				sock = reusedPort.get(k);
				reusedPort.remove(k);
			} else {
				sock = DatagramChannel.open();
				sock.configureBlocking(false);
				newPortThisSecond += 1;
				if (newPortMap.containsKey(ft))
					newPortMap.put(ft, newPortMap.get(ft) + 1);
				else
					newPortMap.put(ft, 1);
			}
			sockMapData dd = new sockMapData();
			dd.num = k;
			dd.createTime = helpFunc.getTime();
			sockMap.put(sock, dd);
			helpFunc.makePack(data, salt);
			sockMap.get(sock).uuid = helpFunc.uuid;
			InetSocketAddress add = new InetSocketAddress(ip, k);
			ByteBuffer bb = ByteBuffer.wrap(helpFunc.con);
			sock.send(bb, add);
			helpFunc.addPackSta(maxSendL, (short) helpFunc.con.length, 0);
			wRaw += helpFunc.con.length;
			if (availPort.containsKey(k))
				availPort.remove(k);
			else
				availPort2.remove(k);
		}
	}
	void deal_timeout() throws Exception {
		ArrayList<DatagramChannel> temp = new ArrayList<DatagramChannel>(sockMap.keySet());
		for (DatagramChannel sock : temp) {
			sockMapData v = sockMap.get(sock);
			if (v.createTime + timeoutTime < helpFunc.getTime()) {
				sock.close();
				int n = sockMap.get(sock).num;
				sockMap.remove(sock);
				if (!retireSock(n))
					availPort.put(n, 0);
				statisOut += 1;
			}
		}
	}


	ArrayList<byte[]> deal_rec(ArrayList<SelectableChannel> l) throws Exception {
		ArrayList<byte[]> re = new ArrayList<byte[]>();
		for (SelectableChannel sock : l) {
			if (sock == outSo)
				continue;
			DatagramChannel sock2 = (DatagramChannel) sock;
			ByteBuffer dd = ByteBuffer.allocate(helpFunc.recLen);
			sock2.receive(dd);

			byte[] j = new byte[dd.position()];
			System.arraycopy(dd.array(), 0, j, 0, dd.position());
			rRaw += j.length;
			helpFunc.checkPackValid2(j, salt);
			byte[] ub = sockMap.get(sock2).uuid;
			int n = sockMap.get(sock2).num;
			double ti = sockMap.get(sock2).createTime;
			if (!Arrays.equals(helpFunc.uuid, ub)) {
				sock2.close();
				sockMap.remove(sock2);
				availPort.put(n, 0);
				statisOut += 1;
				continue;
			} else {
				helpFunc.addPackSta(maxRec, (short) j.length, 0);
				double recT = helpFunc.getTime() - ti;
				if (recT > maxRecTime) {
					maxRecTime = recT;
					if (maxRecTime > timeoutTime)
						maxRecTime = timeoutTime;
				}
				if (recT < minRecTime)
					minRecTime = recT;
				if (retireSock(n))
					sock2.close();
				else {
					availPort2.put(n, 0);
					reusedPort.put(n, sock2);
				}
				sockMap.remove(sock2);
				statisGot += 1;

				re.add(helpFunc.con);
			}
		}
		// helpFunc.print("got num"+Integer.toString(statisGot));
		return re;
	}


	public static void main(String[] args) throws Exception
	{
/*
		SocketChannel so = SocketChannel.open();
		ArrayList<Integer> listenPort = new ArrayList<Integer>();
		for (int i = 50000; i < 50100; i++)
			listenPort.add(i);
		String ss = "salt";
		streamc st = new streamc(listenPort, ss.getBytes(), helpFunc.con_minRate, helpFunc.pushAhead,
				helpFunc.packLimit);
		myThr aa = new myThr();
		aa.st = st;
		aa.setDaemon(true);
		aa.start();
		InetSocketAddress add = new InetSocketAddress("localhost", st.listenPort);
		so.connect(add);
		System.out.println("ok");
		while (true) {
			byte [] sss = new byte[1000];
			ByteBuffer hehe = ByteBuffer.wrap(sss);
			so.write(hehe);
		}
*/
	}


}
